package com.jfree.config;

import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.DocWriteRequest;
import org.elasticsearch.action.DocWriteResponse;
import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.support.IndicesOptions;
import org.elasticsearch.action.support.WriteRequest;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.action.support.replication.ReplicationResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.VersionType;
import org.elasticsearch.rest.RestStatus;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

/**
 * @author wangjian
 * @date 2019/8/26 10:52
 */
public class EsClient {

    private final Logger logger = LoggerFactory.getLogger(EsClient.class);
    private RestHighLevelClient client;

    public EsClient(RestHighLevelClient restHighLevelClient) {
        this.client = restHighLevelClient;
    }

    public CreateIndexResponse createIndex(String indexName,String type,CreateOption option) throws IOException {
        CreateIndexRequest request = new CreateIndexRequest(indexName);
        buildSetting(request);
        buildIndexMapping(request, type);
        //超时,等待所有节点被确认(使用TimeValue方式)
        request.timeout(TimeValue.timeValueMinutes(2));
        //连接master节点的超时时间(使用TimeValue方式)
        request.masterNodeTimeout(TimeValue.timeValueMinutes(1));
        //在创建索引API返回响应之前等待的活动分片副本的数量，以int形式表示。
        request.waitForActiveShards(2);
        final CreateIndexResponse[] res = new CreateIndexResponse[1];
        //request.waitForActiveShards(ActiveShardCount.DEFAULT);//在创建索引API返回响应之前等待的活动分片副本的数量，以ActiveShardCount形式表示。
       if(option.getOption()==CreateOption.SYNC.getOption()) {
           //同步执行
           res[0] = client.indices().create(request, RequestOptions.DEFAULT);
       }else if(option.getOption()==CreateOption.ASYNC.getOption()) {
           client.indices().createAsync(request, RequestOptions.DEFAULT,new ActionListener<CreateIndexResponse>() {
               @Override
               public void onResponse(CreateIndexResponse createIndexResponse) {
                   //如果执行成功，则调用onResponse方法;
                   res[0] =createIndexResponse;
                   logger.info("创建索引成功,indexName:{},type:{}", indexName, type);
               }

               @Override
               public void onFailure(Exception e) {
                   //如果失败，则调用onFailure方法。
                   logger.error("创建索引失败,indexName:{},type:{}", indexName, type);
               }
           });
       }
       CreateIndexResponse response=res[0];
       if(response!=null){
           //指示是否所有节点都已确认请求
           boolean acknowledged=response.isAcknowledged();
           //指示是否在超时之前为索引中的每个分片启动了必需的分片副本数
           boolean shardsAcknowledged = response.isShardsAcknowledged();

       }
       return response;
    }


    public void deleteIndex() throws IOException {
        DeleteIndexRequest request = new DeleteIndexRequest("twitter_two");//指定要删除的索引名称
        //可选参数：
        request.timeout(TimeValue.timeValueMinutes(2)); //设置超时，等待所有节点确认索引删除（使用TimeValue形式）
        // request.timeout("2m"); //设置超时，等待所有节点确认索引删除（使用字符串形式）

        request.masterNodeTimeout(TimeValue.timeValueMinutes(1));////连接master节点的超时时间(使用TimeValue方式)
        // request.masterNodeTimeout("1m");//连接master节点的超时时间(使用字符串方式)

        //设置IndicesOptions控制如何解决不可用的索引以及如何扩展通配符表达式
        request.indicesOptions(IndicesOptions.lenientExpandOpen());

        //同步执行
        AcknowledgedResponse response = client.indices().delete(request, RequestOptions.DEFAULT);

  /*    //异步执行删除索引请求需要将DeleteIndexRequest实例和ActionListener实例传递给异步方法：
        //DeleteIndexResponse的典型监听器如下所示：
        //异步方法不会阻塞并立即返回。
        ActionListener<DeleteIndexResponse> listener = new ActionListener<DeleteIndexResponse>() {
            @Override
            public void onResponse(DeleteIndexResponse deleteIndexResponse) {
                //如果执行成功，则调用onResponse方法;
            }

            @Override
            public void onFailure(Exception e) {
                //如果失败，则调用onFailure方法。
            }
        };
        client.indices().deleteAsync(request, listener);*/

        //Delete Index Response
        //返回的DeleteIndexResponse允许检索有关执行的操作的信息，如下所示：
        boolean acknowledged = response.isAcknowledged();//是否所有节点都已确认请求


        //如果找不到索引，则会抛出ElasticsearchException：
        try {
            request = new DeleteIndexRequest("does_not_exist");
            client.indices().delete(request);
        } catch (ElasticsearchException exception) {
            if (exception.status() == RestStatus.NOT_FOUND) {
                //如果没有找到要删除的索引，要执行某些操作
            }
        }
    }

    public void dataAdd() throws IOException {
        IndexRequest indexRequest1 = new IndexRequest(
                "posts",//索引名称
                "doc",//类型名称
                "1");//文档ID

        //==============================提供文档源========================================
        //方式1：以字符串形式提供
        String jsonString = "{" +
                "\"user\":\"kimchy\"," +
                "\"postDate\":\"2013-01-30\"," +
                "\"message\":\"trying out Elasticsearch\"" +
                "}";
        indexRequest1.source(jsonString, XContentType.JSON);

        //方式2：以Map形式提供
        Map<String, Object> jsonMap = new HashMap<>();
        jsonMap.put("user", "kimchy");
        jsonMap.put("postDate", new Date());
        jsonMap.put("message", "trying out Elasticsearch");
        //Map会自动转换为JSON格式的文档源
        IndexRequest indexRequest2 = new IndexRequest("posts", "doc", "1")
                .source(jsonMap);

        // 方式3：文档源以XContentBuilder对象的形式提供，Elasticsearch内部会帮我们生成JSON内容

        XContentBuilder builder = XContentFactory.jsonBuilder();
        builder.startObject();
        {
            builder.field("user", "kimchy");
            builder.field("postDate", new Date());
            builder.field("message", "trying out Elasticsearch");
        }
        builder.endObject();
        IndexRequest indexRequest3 = new IndexRequest("posts", "doc", "1")
                .source(builder);

        //方式4：以Object key-pairs提供的文档源，它会被转换为JSON格式
        IndexRequest indexRequest4 = new IndexRequest("posts", "doc", "1")
                .source("user", "kimchy",
                        "postDate", new Date(),
                        "message", "trying out Elasticsearch");

        //===============================可选参数start====================================
        indexRequest1.routing("routing");//设置路由值
        indexRequest1.parent("parent");//设置parent值

        //设置超时：等待主分片变得可用的时间
        indexRequest1.timeout(TimeValue.timeValueSeconds(1));//TimeValue方式
        indexRequest1.timeout("1s");//字符串方式

        //刷新策略
        indexRequest1.setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL);//WriteRequest.RefreshPolicy实例方式
        indexRequest1.setRefreshPolicy("wait_for");//字符串方式

        indexRequest1.version(2);//设置版本

        indexRequest1.versionType(VersionType.EXTERNAL);//设置版本类型

        //操作类型
        indexRequest1.opType(DocWriteRequest.OpType.CREATE);//DocWriteRequest.OpType方式
        indexRequest1.opType("create");//字符串方式, 可以是 create 或 update (默认)

        //The name of the ingest pipeline to be executed before indexing the document
        indexRequest1.setPipeline("pipeline");

        //===============================执行====================================
        //同步执行
        IndexResponse indexResponse = client.index(indexRequest1);

        //异步执行
        //IndexResponse 的典型监听器如下所示：
        //异步方法不会阻塞并立即返回。
        ActionListener<IndexResponse> listener = new ActionListener<IndexResponse>() {
            @Override
            public void onResponse(IndexResponse indexResponse) {
                //执行成功时调用。 Response以参数方式提供
            }

            @Override
            public void onFailure(Exception e) {
                //在失败的情况下调用。 引发的异常以参数方式提供
            }
        };
        //异步执行索引请求需要将IndexRequest实例和ActionListener实例传递给异步方法：
        client.indexAsync(indexRequest2, listener);

        //Index Response
        //返回的IndexResponse允许检索有关执行操作的信息，如下所示：
        String index = indexResponse.getIndex();
        String type = indexResponse.getType();
        String id = indexResponse.getId();
        long version = indexResponse.getVersion();
        if (indexResponse.getResult() == DocWriteResponse.Result.CREATED) {
            //处理（如果需要）第一次创建文档的情况
        } else if (indexResponse.getResult() == DocWriteResponse.Result.UPDATED) {
            //处理（如果需要）文档被重写的情况
        }
        ReplicationResponse.ShardInfo shardInfo = indexResponse.getShardInfo();
        if (shardInfo.getTotal() != shardInfo.getSuccessful()) {
            //处理成功分片数量少于总分片数量的情况
        }
        if (shardInfo.getFailed() > 0) {
            for (ReplicationResponse.ShardInfo.Failure failure : shardInfo.getFailures()) {
                String reason = failure.reason();//处理潜在的失败
            }
        }

        //如果存在版本冲突，则会抛出ElasticsearchException：
        IndexRequest request = new IndexRequest("posts", "doc", "1")
                .source("field", "value")
                .version(1);
        try {
            IndexResponse response = client.index(request);
        } catch(ElasticsearchException e) {
            if (e.status() == RestStatus.CONFLICT) {
                //引发的异常表示返回了版本冲突错误
            }
        }

        //如果opType设置为创建但是具有相同索引，类型和ID的文档已存在，则也会发生同样的情况：
        request = new IndexRequest("posts", "doc", "1")
                .source("field", "value")
                .opType(DocWriteRequest.OpType.CREATE);
        try {
            IndexResponse response = client.index(request);
        } catch(ElasticsearchException e) {
            if (e.status() == RestStatus.CONFLICT) {
                //引发的异常表示返回了版本冲突错误
            }
        }
    }


    /**
     * 批量添加  删除
     */
    public void bulkRequest() throws IOException {
        //添加
        BulkRequest request = new BulkRequest();
        request.add(new IndexRequest("posts").id("5")
                .source(XContentType.JSON, "user", "赵琦水岸城1", "name", "赵琦水岸城1"));
        request.add(new IndexRequest("posts").id("7")
                .source(XContentType.JSON, "user", "张三红包2", "name", "马六红包2"));
        request.add(new IndexRequest("posts").id("8")
                .source(XContentType.JSON, "user", "王五红包3", "name", "王五红包3"));
        request.add(new IndexRequest("posts").id("9")
                .source(XContentType.JSON, "user", "王五", "name", "zhangsan红包4"));
        request.add(new IndexRequest("posts").id("10")
                .source(XContentType.JSON, "user", "李四红包5", "name", "李四红包5"));
        request.add(new IndexRequest("posts").id("6")
                .source(XContentType.JSON, "user", "张三红包6", "name", "张三红包6"));
//
//        删除
//        BulkRequest request1 = new BulkRequest();
//        request1.add(new DeleteRequest("posts", "3"));
//        request1.add(new UpdateRequest("posts", "2")
//                .doc(XContentType.JSON,"other", "test"));
//        request1.add(new IndexRequest("posts").id("4")
//                .source(XContentType.JSON,"field", "baz"));

        BulkResponse bulkResponse = client.bulk(request, RequestOptions.DEFAULT);
    }








    //设置分片
    public void buildSetting(CreateIndexRequest request) {
        request.settings(Settings.builder().put("index.number_of_shards", 3)
                .put("index.number_of_replicas", 2));

    }

    //设置index的mapping
    public void buildIndexMapping(CreateIndexRequest request, String type) {
        Map<String, Object> jsonMap = new HashMap<>();
        Map<String, Object> number = new HashMap<>();
        number.put("type", "text");
        Map<String, Object> price = new HashMap<>();
        price.put("type", "float");
        Map<String, Object> title = new HashMap<>();
        title.put("type", "text");
        Map<String, Object> province = new HashMap<>();
        province.put("type", "text");
        Map<String, Object> publishTime = new HashMap<>();
        publishTime.put("type", "date");
        Map<String, Object> properties = new HashMap<>();
        properties.put("number", number);
        properties.put("price", price);
        properties.put("title", title);
        properties.put("province", province);
        properties.put("publishTime", publishTime);
        Map<String, Object> book = new HashMap<>();
        book.put("properties", properties);
        jsonMap.put(type, book);
        request.mapping(type, jsonMap);
    }

    enum   CreateOption{

        SYNC(1,"同步执行"),
        ASYNC(2,"异步执行");

        private int option;
        private String des;

         CreateOption(int option,String des){
            this.option=option;
            this.des=des;
        }

        public int getOption() {
            return option;
        }

        public void setOption(int option) {
            this.option = option;
        }

        public String getDes() {
            return des;
        }

        public void setDes(String des) {
            this.des = des;
        }
    }
}